from decimal import Decimal
from functools import reduce
from typing import *
import asyncio
import aiohttp
import hmac
import hashlib
import base64
import time

import numpy as np
import pandas as pd
from utils.exchange.baseExchange import BaseExchange
from config import PROXY_CONF
from utils.error import RealExchangeError, PreparationError
from functools import wraps


def format_interval(interval: int) -> str:
    """"""
    day = interval // 86400
    hour = interval // 3600
    minute = interval // 60
    if day > 0:
        return f'{day}d'
    if hour > 0:
        return f'{hour}h'
    if minute > 0:
        return f'{minute}m'

    raise PreparationError('间隔不得小于1分钟')


def async_trans(func: Callable[..., Coroutine]) -> Callable[..., Union[Coroutine, Any]]:
    """
    根据配置选择是否将异步转同步
    :param func:
    :return:
    """

    @wraps(func)
    def wrapper(ex: "BinanceBase", *args, **kwargs) -> Union[Any, Coroutine]:
        if ex.async_:
            return asyncio.create_task(func(ex, *args, **kwargs))
        return asyncio.run(func(ex, *args, **kwargs))

    return wrapper


class BinanceBase(BaseExchange):
    spot_api: str = 'https://api3.binance.com'
    swap_api: str = 'https://fapi.binance.com'

    def __init__(self, api_key: str = None, api_secret: str = None, recv_window: int = 5000, async_: bool = False):
        """

        :param api_key:
        :param api_secret:
        :param recv_window:
        :param async_: 是否为异步模式
        """
        super().__init__()
        self.api_key = api_key
        self.api_secret = api_secret
        self.recv_window = recv_window
        self.async_ = async_

    def _sign(self, message: str) -> str:
        """
        请求签名
        :param message:
        :return:
        """
        h = hmac.new(self.api_secret.encode(), message.encode(), digestmod=hashlib.sha256).digest()
        return base64.b16encode(h).decode().lower()

    def _query_format(self, data: Dict[str, Any], sign: bool = True) -> str:
        """
        请求参数格式化签名
        :param data:
        :param sign: 是否加密
        :return:
        """

        _query = '&'.join([f'{key}={data[key]}' for key in data.keys()]) if len(data) else ''

        if not sign:
            return f'?{_query}' if _query else _query

        if self.api_key is None or self.api_secret is None:
            raise RealExchangeError('访问加密接口，请提供aipKey和secret')
        signature = self._sign(_query)
        return '?' + _query + f'&signature={signature}'

    async def _fetch(self, session: aiohttp.ClientSession, url: str, params: Dict[str, Any], method: str = 'GET',
                     spot: bool = True, sign: bool = True, async_: bool = False) -> Union[Any, Tuple[bool, Any]]:
        """
        发送请求到币安
        :param session: 异步session
        :param url:
        :param params: 参数
        :param method: 请求方式
        :param spot:
        :param sign:
        :param async_: 是否为多任务模式
        :return:
        """
        if sign:
            params['recvWindow'] = self.recv_window
            params['timestamp'] = int(time.time()) * 1000 if params.get('timestamp') is None else params['timestamp']

        url = (self.spot_api if spot else self.swap_api) + url + self._query_format(params, sign)
        headers = {'X-MBX-APIKEY': self.api_key} if sign else {}
        resp = await session.request(method, url, headers=headers, proxy=PROXY_CONF.get('https'))
        if resp.status == 200:
            return (True, await resp.json()) if async_ else await resp.json()
        if resp.status == 429:
            raise RealExchangeError(f'触发交易所访问限制{await resp.text()}')
        if resp.status != 200:
            if async_:
                return False, await resp.json()
            raise RealExchangeError(f'请求交易所错误：{resp.status}, {url}, {await resp.text()}')

    async def _batch_fetch(self, url: str, params: Union[dict, List[dict]], method: str = 'GET',
                           spot: bool = True, sign: bool = True, async_: bool = False) \
            -> Union[Any, List[Tuple[bool, Any]]]:
        """
        合并执行请求
        :param url: 路由
        :param params: 请求参数
        :param method: 请求方式
        :param spot: 是否现货接口
        :param sign: 是否加密
        :param async_: 是否作为子任务
        :return:
        """
        async with aiohttp.ClientSession() as session:
            if not isinstance(params, list):
                return await self._fetch(session, url, params, method, spot, sign, async_=async_)
            # 多任务模式
            resp = await asyncio.gather(
                *[self._fetch(session, url, param, method, spot, sign, async_=True) for param in params])
            return list(resp)

    @async_trans
    async def kline(self, spot: bool = True, symbols: Sequence[str] = None,
                    start_time: int = None, limit: int = 1, interval: int = 3600
                    ) -> pd.DataFrame:
        """
        获取k线数据 weight 1-10
        :param spot:
        :param symbols:
        :param start_time:
        :param limit:
        :param interval:
        :return:
        """
        _route: str = '/api/v3/klines' if spot else '/fapi/v1/klines'
        if symbols is None:
            symbols = (await self.spot_exchange_info).keys() if spot else (await self.swap_exchange_info).keys()
            symbols = list(symbols)
        if start_time is None:
            start_time = int(time.time()) // interval * interval - interval * limit
        resp = await self._batch_fetch(_route, [{
            'symbol': x, 'interval': format_interval(interval), 'startTime': start_time * 1000, 'limit': limit
        } for x in symbols], spot=spot, sign=False)
        dfs = [pd.DataFrame(x[1], columns=['open_time', 'open', 'high', 'low', 'close', 'volume',
                                           'close_time', 'quote_asset_volume', 'number_of_trades',
                                           'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume',
                                           'ignore'], dtype=float) for x in resp if x[0]]
        for i, df in enumerate(dfs):
            df['open_time'] = df['open_time'].astype(np.int64)
            df['close_time'] = df['close_time'].astype(np.int64)
            df['open_time'] = (df['open_time'] / 1000).round() * 1000
            df['symbol'] = symbols[i]
            df['trade_type'] = 0 if spot else 1
        return pd.concat(dfs, ignore_index=True)

    @async_trans
    async def funding_rate(self, start_time: int, symbols: Sequence[str],
                           end_time: int = None) -> pd.DataFrame:
        """
        获取资金费率 币安接口逻辑存在问题，有数据缺失风险 weight 0
        :param start_time:
        :param symbols:
        :param end_time:
        :return: col为 symbol,fundingTime,fundingRate,markPrice
        """
        _route: str = '/fapi/v1/fundingRate'
        params = [{'startTime': start_time * 1000, 'symbol': x, 'limit': 1000,
                   'endTime': end_time * 1000 if end_time is not None else int(time.time() * 1000)} for x in symbols]

        resp: List[Tuple[bool, Any]] = await self._batch_fetch(_route, params, spot=False, sign=False)
        resp = reduce(lambda x, y: x + y, [x[1] for x in resp if x[0]])
        if len(resp) == 0:
            return pd.DataFrame(columns=['symbol', 'fundingTime', 'fundingRate', 'markPrice'])
        data = pd.DataFrame(resp)
        data['fundingRate'] = data['fundingRate'].astype(float)
        return data

    @async_trans
    async def multi_asset_type(self, type_: bool = False):
        """
        更改联合保证金模式
        :param type_: 是否开启
        :return:
        """
        _route: str = '/fapi/v1/multiAssetsMargin'
        resp: Dict[str, Any] = await (
            self._batch_fetch(_route, {'multiAssetsMargin': type_}, method='POST', spot=False, sign=False)
        )
        return resp

    @async_trans
    async def margin_type(self, symbol: List[str], update: str = 'ISOLATED'):
        """
        调整仓位模式 ISOLATED(逐仓), CROSSED(全仓)
        权重： 1/symbol
        :param symbol:
        :param update:
        :return:
        """
        _route: str = '/fapi/v1/marginType'
        resp = await (
            self._batch_fetch(_route, [{'symbol': x, 'marginType': update} for x in symbol], method='POST', spot=False))
        for res in resp:
            if not res[0] and res[1]['code'] == -4168:
                try:
                    await self.multi_asset_type(False)
                except Exception as e:
                    raise RealExchangeError(f'修改联合保证金失败：{e.__str__()}')
            if not res[0] and res[1]['code'] != -4046:
                raise RealExchangeError(f'修改仓位模式错误：{res[1]["msg"]}')
        return resp

    @property
    @async_trans
    async def key_permission(self) -> Dict[str, Any]:
        """
        api key权限信息
        :return:
        """
        resp = await self._batch_fetch('/sapi/v1/account/apiRestrictions', {})
        return resp

    @async_trans
    async def swap_position(self) -> List[Dict[str, Any]]:
        """
        合约持仓 双向或者单向 weight 5
        :return:
        """
        _route: str = '/fapi/v2/positionRisk'
        result: Any = await self._batch_fetch(_route, {}, spot=False)

        return [{'symbol': x['symbol'],
                 'positionAmt': Decimal(x['positionAmt']),
                 'unRealizedProfit': Decimal(x['unRealizedProfit']),
                 'entryPrice': Decimal(x['entryPrice']),
                 'leverage': Decimal(x['leverage']),
                 } for x in result if Decimal(x['positionAmt']) != 0]

    @async_trans
    async def asset_balance(self) -> Dict[str, Decimal]:
        _route: str = '/sapi/v1/asset/get-funding-asset'
        result: Any = await self._batch_fetch(_route, {}, method='POST')
        return {x['asset']: Decimal(x['free']) for x in result}

    @async_trans
    async def swap_balance(self) -> Dict[str, Dict[str, Decimal]]:
        """
        合约资产
        weight: 5
        :return:
        """
        _route: str = '/fapi/v2/balance'
        result: Any = await self._batch_fetch(_route, {}, spot=False)
        return {x['asset']: {'balance': Decimal(x['balance']),
                             'maxWithdrawAmount': Decimal(x['maxWithdrawAmount']),
                             } for x in result if
                Decimal(x['balance']) != 0}

    @async_trans
    async def spot_balance(self) -> Dict[str, Decimal]:
        """
        weight 5
        现货资产
        :return:
        """
        _route: str = '/sapi/v1/asset/getUserAsset'
        result: Any = await self._batch_fetch(_route, {})
        return {x['asset']: Decimal(x['free']) for x in result}

    @async_trans
    async def spot_trade(self, trades: Union[List[Dict[str, Any]], Dict[str, Any]]) -> List[Tuple[bool, Any]]:
        """
        批量现货交易
        :param trades:
        :return:
        """
        _route: str = '/api/v3/order'
        result: List[Tuple[bool, Any]] = await self._batch_fetch(_route, trades, method='POST')
        return result

    @async_trans
    async def swap_trade(self, trades: Union[List[Dict[str, Any]], Dict[str, Any]]) -> List[Tuple[bool, Any]]:
        """
        批量合约交易
        :param trades:
        :return:
        """
        _route: str = '/fapi/v1/order'
        result: List[Tuple[bool, Any]] = await self._batch_fetch(_route, trades, method='POST', spot=False)
        return result

    @async_trans
    async def transfer(self, orders: List[Dict[str, Any]]) -> List[Tuple[bool, Any]]:
        """
        [{'type': str, 'asset': str, 'amount': Decimal}]
        MAIN_UMFUTURE 现货钱包转向U本位合约钱包
        UMFUTURE_MAIN U本位合约钱包转向现货钱包
        MAIN_FUNDING 现货钱包转向资金钱包
        FUNDING_MAIN 资金钱包转向现货钱包
        FUNDING_UMFUTURE 资金钱包转向U本位合约钱包
        UMFUTURE_FUNDING U本位合约钱包转向资金钱包
        权重(UID)): 900
        :return:
        """
        _param = [{'type': x['type'], 'asset': x['asset'], 'amount': x['amount']} for x in orders if x['amount'] > 0]
        resp = await self._batch_fetch('/sapi/v1/asset/transfer', _param, method='POST')
        return resp

    @async_trans
    async def transfer_history(self, orders: List[Dict[str, Any]]) -> List[Tuple[bool, Any]]:
        """
        [{'type': str, 'startTime': int}]
        MAIN_UMFUTURE 现货钱包转向U本位合约钱包
        UMFUTURE_MAIN U本位合约钱包转向现货钱包
        MAIN_FUNDING 现货钱包转向资金钱包
        FUNDING_MAIN 资金钱包转向现货钱包
        FUNDING_UMFUTURE 资金钱包转向U本位合约钱包
        UMFUTURE_FUNDING U本位合约钱包转向资金钱包
        :return:
        """
        _param = []
        for item in orders:
            param = {'type': item['type']}
            if 'startTime' in item.keys():
                param['startTime'] = item['startTime']

            _param.append(param)

        resp = await self._batch_fetch('/sapi/v1/asset/transfer', _param)
        return resp

    @property
    @async_trans
    async def spot_exchange_info(self) -> Dict[str, Dict[str, Decimal]]:
        """
        交易对规则信息 weight 20
        :return:
        """
        resp: Dict[str, Any] = await self._batch_fetch('/api/v3/exchangeInfo', {}, sign=False)
        symbols = resp['symbols']

        info: Dict[str, Dict[str, Any]] = {}
        for symbol in symbols:
            if symbol['quoteAsset'] != 'USDT':
                continue
            temp: Dict[str, Any] = {}
            for f in symbol['filters']:
                # 最小下单量
                if f['filterType'] == 'PRICE_FILTER':
                    temp['tickSize'] = Decimal(f['tickSize']).normalize()
                # 下单精度
                elif f['filterType'] == 'LOT_SIZE':
                    temp['minQty'] = Decimal(f['minQty']).normalize()

                elif f['filterType'] == 'NOTIONAL':
                    temp['minNotional'] = Decimal(f['minNotional']).normalize()
                else:
                    continue

                info[symbol['symbol']] = temp
        return info

    @property
    @async_trans
    async def swap_exchange_info(self) -> Dict[str, Dict[str, Decimal]]:
        """

        :return:
        """
        resp: Dict[str, Any] = await self._batch_fetch('/fapi/v1/exchangeInfo', {}, sign=False, spot=False)
        symbols = resp['symbols']

        info: Dict[str, Dict[str, Any]] = {}
        for symbol in symbols:
            if symbol['quoteAsset'] != 'USDT':
                continue
            temp: Dict[str, Any] = {}
            for f in symbol['filters']:
                if f['filterType'] == 'PRICE_FILTER':
                    temp['tickSize'] = Decimal(f['tickSize']).normalize()
                # 最小下单
                elif f['filterType'] == 'LOT_SIZE':
                    temp['minQty'] = Decimal(f['minQty']).normalize()

                elif f['filterType'] == 'MIN_NOTIONAL':
                    temp['minNotional'] = Decimal(f['notional']).normalize()
                else:
                    continue

                info[symbol['symbol']] = temp
        return info

    @property
    @async_trans
    async def spot_symbol_prices(self) -> Dict[str, Decimal]:
        resp: List[Dict[str, Any]] = await self._batch_fetch('/api/v3/ticker/price', {}, sign=False)
        return {x['symbol']: Decimal(x['price']) for x in resp}

    @property
    @async_trans
    async def swap_symbol_prices(self) -> Dict[str, Decimal]:
        resp: List[Dict[str, Any]] = await self._batch_fetch('/fapi/v1/ticker/price',
                                                             {}, sign=False, spot=False)
        return {x['symbol']: Decimal(x['price']) for x in resp}

    @async_trans
    async def swap_position_side(self, update_to: bool = None) -> bool:
        """
        合约持仓方向
        weight 查30 改1
        :param update_to: False 为单向持仓
        :return:
        """
        if update_to is None:
            resp: dict = await self._batch_fetch('/fapi/v1/positionSide/dual', {}, spot=False)
            return resp['dualSidePosition']
        await self._batch_fetch('/fapi/v1/positionSide/dual',
                                {'dualSidePosition': update_to}, spot=False, method='POST')

    @async_trans
    async def swap_leverage(self, leverage: List[int], symbol: List[str]) -> Dict[str, Any]:
        """
        更改币种默认杠杆
        :param leverage:
        :param symbol:
        :return:
        """
        _route = '/fapi/v1/leverage'
        resp: Dict[str, Any] = await self._batch_fetch(_route,
                                                       [{'symbol': x, 'leverage': y} for x, y in zip(symbol, leverage)],
                                                       spot=False, method='POST')
        return {symbol[i]: x[0] for i, x in enumerate(resp)}

    @async_trans
    async def get_swap_orders(self, symbols: List[str] = None):
        """
        获取合约挂单
        :param symbols:
        :return:
        """
        _route = '/fapi/v1/openOrders'
        if symbols is None:
            resp: List[dict] = await self._batch_fetch(_route, {}, spot=False)
            return resp
        resp: List[Tuple[bool, List[dict]]] = await self._batch_fetch(_route, [{'symbol': x} for x in symbols],
                                                                      spot=False)
        return resp

    @async_trans
    async def delete_swap_orders(self, symbol: str, order_ids: List[str] = None):
        """
        取消挂单
        weight 1/order
        :param symbol:
        :param order_ids:
        :return:
        """
        _route = '/fapi/v1/allOpenOrders' if order_ids is None else '/fapi/v1/order'
        params = {'symbol': symbol} if order_ids is None else [{'symbol': symbol, 'orderId': x} for x in order_ids]
        return await self._batch_fetch(_route, params=params, method='DELETE', spot=False)

    @async_trans
    async def get_convertable_asset(self) -> Dict[str, Any]:
        """
        获取可转换为BNB的现货资产
        :return:
        """
        _route = '/sapi/v1/asset/dust-btc'
        params = {}
        data: dict = await self._batch_fetch(_route, params, method='POST')
        return {x['asset']: x['amountFree'] for x in data['details']}

    @async_trans
    async def convert_asset(self, assets: List[str]):
        """
        将现货资产转换为BNB
        :param assets:
        :return:
        """
        _route = '/sapi/v1/asset/dust'
        params = {'asset': ','.join(assets)} if assets else {}
        return await self._batch_fetch(_route, params, method='POST')

    def __str__(self):
        return f'Binance: {self.api_key}'


if __name__ == '__main__':
    import yaml
    import os
    from config import ROOT_DIR

    with open(os.path.join(ROOT_DIR, 'account.yaml'), 'r', encoding='utf-8') as f:
        acs = yaml.safe_load(f)
    # b = BinanceBase()
    b = BinanceBase(acs['accounts']['masterAccount']['apiKey'], acs['accounts']['masterAccount']['secretKey'])

    # b.swap_leverage([2], ['BTCUSDT'])
    # # print(b.swap_exchange_info['ETHUSDT'])
    # resp = b.swap_trade(
    #     [{'positionSide': 'BOTH', 'symbol': 'BTCUSDT',
    #         'type': 'MARKET', 'quantity': '0.015', 'side': 'SELL', 'reduceOnly': True,},
    #      {'positionSide': 'BOTH', 'symbol': 'BNBUSDT', 'type': 'MARKET', 'quantity': '0.01', 'side': 'SELL',
    #       'reduceOnly': True,},
    #      ])
    # print(b.transfer([{'type': 'UMFUTURE_MAIN', 'asset': 'USDT', 'amount': Decimal('621.73223781')}]))
    print(b.swap_position())
    # print(resp)
    # print(b.swap_position())
    # print(b.transfer([{'type':'MAIN_UMFUTURE', 'asset': 'USDT', 'amount': Decimal(480)}]))
    # b.margin_type(['BTCUSDT'], )

